package com.geek.netty.outbound.httpclient4;

import com.geek.netty.filter.HeaderHttpResponseFilter;
import com.geek.netty.filter.HttpRequestFilter;
import com.geek.netty.filter.HttpResponseFilter;
import com.geek.netty.router.HttpEndPointRouter;
import com.geek.netty.router.RandomHttpEndPointRouter;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpUtil;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.protocol.HTTP;
import org.apache.http.util.EntityUtils;

import java.util.List;
import java.util.concurrent.*;

import static io.netty.handler.codec.http.HttpResponseStatus.NO_CONTENT;
import static io.netty.handler.codec.http.HttpResponseStatus.OK;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
import static java.util.stream.Collectors.toList;

/**
 * @program geek
 * @description: handler
 * @author: chanzhugang
 * @create: 2021/04/04 12:12
 */
public class HttpOutboundHandler {

    private CloseableHttpAsyncClient httpClient;

    private ExecutorService proxyService;

    private List<String> backendUrls;

    HttpResponseFilter filter = new HeaderHttpResponseFilter();

    HttpEndPointRouter router = new RandomHttpEndPointRouter();

    /**
     * 定义线程池、httpClient
     */
    public HttpOutboundHandler(List<String> backends) {
        this.backendUrls = backends.stream().map(this::formatUrl).collect(toList());

        // 获取当前CPU核心数
        int cores = Runtime.getRuntime().availableProcessors();
        long keepAliveTime = 1000;
        int queueSize = 2048;

        RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy();
        proxyService = new ThreadPoolExecutor(cores, cores,
                keepAliveTime, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(queueSize),
                new NamedThreadFactory("proxyService"), handler);

        IOReactorConfig ioConfig = IOReactorConfig.custom()
                .setConnectTimeout(1000)
                .setSoTimeout(1000)
                .setIoThreadCount(cores)
                .setRcvBufSize(32 * 1024)
                .build();

        httpClient = HttpAsyncClients.custom().setMaxConnTotal(40)
                .setMaxConnPerRoute(8)
                .setDefaultIOReactorConfig(ioConfig)
                .setKeepAliveStrategy((response, context) -> 6000)
                .build();

        httpClient.start();
    }

    private String formatUrl(String backend) {
        return backend.endsWith("/") ? backend.substring(0, backend.length() - 1) : backend;
    }


    public void handle(FullHttpRequest fullHttpRequest, ChannelHandlerContext cxt, HttpRequestFilter filter) {
        // 随机路由
        String backendUrl = router.route(this.backendUrls);
        String url = backendUrl + fullHttpRequest.uri();
        // 添加请求头 （过滤）
        filter.filter(fullHttpRequest, cxt);

        // 线程池 处理服务端
        proxyService.submit(() -> fetchGet(fullHttpRequest, cxt, url));
    }

    private void fetchGet(FullHttpRequest fullHttpRequest, ChannelHandlerContext cxt, String url) {
        HttpGet httpGet = new HttpGet(url);
        httpGet.setHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE);
        httpGet.setHeader("nettyHeader", fullHttpRequest.headers().get("nettyHeader"));

        System.out.println(fullHttpRequest.headers().get("nettyHeader"));
        // 提交任务
        httpClient.execute(httpGet, new FutureCallback<HttpResponse>() {
            @Override
            public void completed(HttpResponse response) {
                try {
                    handResponse(fullHttpRequest, cxt, response);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {

                }
            }

            @Override
            public void failed(Exception e) {
                httpGet.abort();
                e.printStackTrace();
            }

            @Override
            public void cancelled() {
                httpGet.abort();
            }
        });


    }

    private void handResponse(FullHttpRequest fullHttpRequest, ChannelHandlerContext cxt, HttpResponse endPointResponse) {
        FullHttpResponse response = null;
        try {
            byte[] body = EntityUtils.toByteArray(endPointResponse.getEntity());
            response = new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.wrappedBuffer(body));
            response.headers().set("Content-type", "application/json");
            response.headers().setInt("Content-Length", Integer.parseInt(endPointResponse.getFirstHeader("Content-Length").getValue()));
            // 添加响应头 （过滤）
            filter.filter(response);

        } catch (Exception e) {
            e.printStackTrace();
            response = new DefaultFullHttpResponse(HTTP_1_1, NO_CONTENT);
            exceptionCaught(cxt, e);
        } finally {
            if (null != fullHttpRequest) {
                if (!HttpUtil.isKeepAlive(fullHttpRequest)) {
                    cxt.write(response).addListener(ChannelFutureListener.CLOSE);
                } else {
                    cxt.write(response);
                }
            }
            cxt.flush();
        }

    }

    private void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}